package com.hibase.core.mq.redis.consumer;

import cn.hutool.core.util.ObjectUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.TypeReference;
import com.hibase.common.constant.RedisKeyConstant;
import com.hibase.core.mq.annotation.HibaseQueueConsumer;
import com.hibase.core.mq.consumer.IQueueConsumer;
import com.hibase.core.redis.util.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

import javax.annotation.PostConstruct;
import java.util.concurrent.Executors;

/**
 * @author 杨斌冰-工具组-技术中心
 *         <p>
 *         2018/3/1 15:31
 */
@Slf4j
public abstract class AbstractRedisQueueConsumer<T> implements IQueueConsumer {

    @Autowired
    private RedisUtil redisUtil;

    private String queueName;

    public abstract void doConsume(T message);

    @Override
    public void consume() throws Exception{
        if(this.getClass().isAnnotationPresent(HibaseQueueConsumer.class)){
            HibaseQueueConsumer redisQueueConsumer = this.getClass().getAnnotation(HibaseQueueConsumer.class);
            queueName = StrUtil.isNotBlank(redisQueueConsumer.value()) ? redisQueueConsumer.value() : redisQueueConsumer.queueName();
            if(StrUtil.isBlank(queueName)){
                throw new Exception("Redis Queue Consumer Java Config Error.");
            }
            while (true){
                Object message = redisUtil.lRightPop(RedisKeyConstant.QUEUE_PREFIX_KEY+queueName);
                Thread.sleep(100);
                if(ObjectUtil.isNotEmpty(message)){
                    this.doConsume(JSON.parseObject((String) message,new TypeReference<T>(){}));
                }
            }
        }else{
            throw new Exception("Redis Queue Consumer Java Config Error.");
        }
    }

    @PostConstruct
    public void init(){
        log.info("The Consumer in Queue [{}] Started.",this.getClass().getName());
        Executors.newSingleThreadExecutor().submit(()->{
            try {
                this.consume();
            } catch (Exception e) {
                log.error("The Consumer in Queue [" + this.queueName + "] Error.",e);
            }
        });
    }
}
